package com.winit.schedule.job.tracker.schedule;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import com.winit.schedule.core.model.AssignedTask;
import com.winit.schedule.core.model.TaskID;
import com.winit.schedule.core.repository.constants.TaskStatus;
import com.winit.schedule.core.repository.entity.TaskEntity;
import com.winit.schedule.core.repository.entity.TaskItemEntity;
import com.winit.schedule.core.repository.entity.TaskRunnerEntity;
import com.winit.schedule.core.repository.entity.TaskTrackerEntity;
import com.winit.schedule.core.task.ScheduleTaskDefine;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.core.utils.CronExpression;
import com.winit.schedule.job.tracker.JobTrackerContext;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;

/**
 * 任务调度器,将单个任务的所有任务项均衡的分配到每台TaskTracker上执行;
 * TaskTracker断开或者新增TaskTracker将会重新计算并分配任务项至每个TaskTracker
 * 
 * @author jianke.zhang 2015-4-16 下午3:08:52
 * @since 1.0
 */
public class EqualizerTaskManager implements TaskScheduleManager {

    private static final transient Logger logger = LoggerFactory.getLogger(EqualizerTaskManager.class);

    private JobTrackerContext             jobTrackerContext;

    private ScheduleDataManager           scheduleDataManager;

    public EqualizerTaskManager(JobTrackerContext jobTrackerContext){
        this.jobTrackerContext = jobTrackerContext;
        this.scheduleDataManager = jobTrackerContext.getScheduleDataManager();
    }

    @Override
    public List<AssignedTask> getExecutableTask(TaskTrackerEntity taskTracker, List<TaskRunnerEntity> runners)
                                                                                                              throws Exception {

        List<AssignedTask> executableTasks = new ArrayList<AssignedTask>();
        boolean hasBlackList = scheduleDataManager.containBlackList(taskTracker.getIp());

        if (!hasBlackList) {
            Date currentDate = Clock.getCurrentDate();
            List<AssignedTask> tasks = getRequestTask(taskTracker, runners);
            TaskEntity taskEntity = null;
            TaskItemEntity taskItem = null;
            for (AssignedTask task : tasks) {
                String taskId = task.getTaskId().getTaskId();
                taskEntity = scheduleDataManager.getTaskEntityById(taskId);
                if (null != taskEntity && TaskStatus.START.equals(taskEntity.getTaskStatus())
                    && allowIpToTaskTracker(taskEntity, taskTracker)) {
                    copyProperties(taskEntity, task);

                    // 计算运行时间
                    computerRunTime(currentDate, task);

                    if (task.getCurrentFetchNum() > 0) {
                        // 不采用本地同步；采用乐观锁，根据数据库update写锁操作de影响行数来获取任务项,
                        // 多JobTracker可以并行。
                        // String lock = String.valueOf(taskId);
                        // synchronized (lock) {
                        taskItem = scheduleDataManager.findAndFetchTaskItemByTaskId(taskTracker.getTaskTrackerUuid(),
                                taskId,
                                task.getPermitRunStartTime(),
                                task.getCurrentFetchNum());
                        // }
                    }

                    // 如果抓取到可执行任务项
                    if (null != taskItem && task.getCurrentFetchNum() > 0) {
                        if (logger.isDebugEnabled()) {
                            logger.debug("Get to the task item : " + taskItem.getTaskItemId() + ", for "
                                         + taskTracker.getTaskTrackerUuid());
                        }
                        task.getTaskItems().add(taskItem);
                        task.setPermitRunStartTime(taskItem.getNextRunTime());
                        executableTasks.add(task);
                    } else if (task.getCurrentFetchNum() < 0) {
                        executableTasks.add(task);
                    }
                } else {
                    if (logger.isDebugEnabled()) {
                        logger.debug("No query task : " + taskId + ", for " + taskTracker.getTaskTrackerUuid());
                    }
                }
            }
        }

        return executableTasks;
    }

    /**
     * 检查是否限定任务执行的IP地址
     * 
     * @param taskEntity
     * @param taskTracker
     * @return
     */
    private boolean allowIpToTaskTracker(TaskEntity taskEntity, TaskTrackerEntity taskTracker) {

        String ipstr = StringUtils.trimAllWhitespace(taskEntity.getIpWhiteList());
        if (null == ipstr || "".equals(ipstr.trim())) {
            return true;
        }

        String[] ipArr = StringUtils.delimitedListToStringArray(ipstr, ",");
        if (null != ipArr) {
            for (String ip : ipArr) {
                if (taskTracker.getIp().equals(ip)) {
                    return true;
                }
            }
        }
        if (logger.isDebugEnabled()) {
            logger.debug(String.format("The IP[%s] address is being task[%s] constraints",
                    taskTracker.getIp(),
                    taskEntity.getTaskId().getTaskId()));
        }
        return false;
    }

    /**
     * 计算任务允许运行的开始时间和结束时间
     * 
     * @param currentDate
     * @param task
     * @throws Exception
     */
    private void computerRunTime(Date currentDate, AssignedTask task) throws Exception {

        if (null != task.getRunStartTime()) {
            CronExpression startTime = new CronExpression(task.getRunStartTime());
            Date firstStartTime = startTime.getNextValidTimeAfter(currentDate);
            long permitRunEndTime = 0;
            if (null != task.getRunEndTime() && !task.getRunEndTime().trim().equals("")) {
                String tmpEndStr = task.getRunEndTime();
                CronExpression cexpEnd = new CronExpression(tmpEndStr);
                Date firstEndTime = cexpEnd.getNextValidTimeAfter(firstStartTime);
                Date nowEndTime = cexpEnd.getNextValidTimeAfter(currentDate);
                if (!nowEndTime.equals(firstEndTime) && currentDate.before(nowEndTime)) {
                    firstEndTime = nowEndTime;
                }
                permitRunEndTime = firstEndTime.getTime();
            }

            if (null != firstStartTime) {
                task.setPermitRunStartTime(firstStartTime.getTime());
            } else {
                task.setPermitRunStartTime(-1);
            }
            task.setPermitRunEndTime(permitRunEndTime);
        } else {
            task.setPermitRunStartTime(Clock.getSystemTime());
        }

    }

    /**
     * 获取taskTracker已定义且还未执行的任务
     * 
     * @param taskTracker
     * @param runners
     * @return
     */
    private List<AssignedTask> getRequestTask(TaskTrackerEntity taskTracker, List<TaskRunnerEntity> runners) {
        List<AssignedTask> tasks = new ArrayList<AssignedTask>();
        AssignedTask task = null;

        Map<String, Integer> task2RunnerMapper = task2RunnerMapper(taskTracker.getTaskTrackerUuid(), runners);

        for (ScheduleTaskDefine define : taskTracker.getTaskDefines()) {
            if (!define.isExclude()) {

                int allowTotalFetchNum = (int) scheduleDataManager.getTaskItemNumberOfPerTasktracker(define.getTaskId());

                Integer count = task2RunnerMapper.get(define.getTaskId());
                int currentRunNum = 0;
                if (null != count) {
                    currentRunNum = count.intValue();
                }

                if (allowTotalFetchNum != currentRunNum) {
                    if (allowTotalFetchNum > currentRunNum) {
                        task = new AssignedTask();
                        task.setAllowTotalFetchNum(allowTotalFetchNum);
                        task.setCurrentFetchNum(1);
                        task.setTaskDefine(define);
                        task.setTaskId(new TaskID(null, define.getTaskId()));
                        tasks.add(task);
                    } else {
                        task = new AssignedTask();
                        task.setAllowTotalFetchNum(allowTotalFetchNum);
                        task.setCurrentFetchNum(allowTotalFetchNum - currentRunNum);
                        task.setTaskDefine(define);
                        task.setTaskId(new TaskID(null, define.getTaskId()));
                        tasks.add(task);
                    }
                }

            }

        }

        return tasks;
    }

    private Map<String, Integer> task2RunnerMapper(String taskTrackerUuid, List<TaskRunnerEntity> runners) {
        Map<String, Integer> mapper = new HashMap<String, Integer>();

        if (null != runners) {
            for (TaskRunnerEntity runner : runners) {
                if (runner.getTaskTrackerUuid().equals(taskTrackerUuid)) {
                    Integer count = mapper.get(runner.getTaskId().getTaskId());
                    if (null == count) {
                        count = new Integer(0);
                    }
                    mapper.put(runner.getTaskId().getTaskId(), count.intValue() + 1);
                }
            }
        }

        return mapper;
    }

    private void copyProperties(TaskEntity source, AssignedTask target) {
        if (null != source && null != target) {
            target.setTaskId(source.getTaskId());
            target.setTaskGroup(source.getTaskGroup());
            target.setTaskName(source.getTaskName());
            target.setInterfaceType(source.getInterfaceType());
            target.setRunStartTime(source.getRunStartTime());
            target.setRunEndTime(source.getRunEndTime());
            target.setMaxThreads(source.getMaxThreads());
            target.setThreadsOfRegion(source.getThreadsOfRegion());
            target.setTaskItem(source.getTaskItem());
            target.setTaskParameter(source.getTaskParameter());
            target.setSleepTimeInterval(source.getSleepTimeInterval());
            target.setSleepTimeNoData(source.getSleepTimeNoData());
            target.setPriority(source.getPriority());
            target.setBatchDealNum(source.getBatchDealNum());
            target.setEachFetchDataNum(source.getEachFetchDataNum());
        }
    }

    public JobTrackerContext getJobTrackerContext() {
        return jobTrackerContext;
    }

}
